package com.atguigu.edu.app.dws;

import com.atguigu.edu.app.BaseSQLApp;
import com.atguigu.edu.bean.KeywordBean;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.function.KwSplit;
import com.atguigu.edu.util.FlinkSinkUtil;
import com.atguigu.edu.util.SQLUtil;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new DwsTrafficKeywordPageViewWindow().init(4001,2,
                "DwsTrafficKeywordPageViewWindow");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //建表与dwd的page_log关联
        tEnv.executeSql("create table page_log( " +
                " item string, " +
                " last_page_id string, " +
                " item_type string, " +
                " ts bigint, " +
                " et as to_timestamp_ltz(ts,3), " +
                " watermark for et as et - interval '3' second " +
                " ) "+ SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE,"DwsTrafficKeywordPageViewWindow"));
        //过滤出搜索关键词的记录
        Table itemTable = tEnv.sqlQuery("select " +
                " item, " +
                " et " +
                " from page_log " +
                " where item is not null " +
                " and item_type = 'keyword' ");
        tEnv.createTemporaryView("item_table",itemTable);
        //创建自定义临时函数
        tEnv.createTemporaryFunction("kw_split", KwSplit.class);
        //使用自定义函数对搜索关键词进行分词
//        Table kwTable = tEnv.sqlQuery(" select " +
//                " kw, " +
//                " et " +
//                " from item_table " +
//                " join lateral table( kw_split(item) ) on true");
//        tEnv.createTemporaryView("kw_table",kwTable);
        //用tvf开窗聚合
        Table result = tEnv.sqlQuery(" select " +
                "date_format(window_start, 'yyyy-MM-dd HH:mm:ss') stt, " +
                "date_format(window_end, 'yyyy-MM-dd HH:mm:ss') edt, " +
                "item keyword, " +
                " count(*) keyword_count, " +
                " unix_timestamp()*1000 as ts " +
                " from table (tumble( table item_table, descriptor(et),interval '5' second ) )" +
                " group by window_start, window_end,item");
        SingleOutputStreamOperator<KeywordBean> resultStream = tEnv.toRetractStream(result, KeywordBean.class)
                .filter(f -> f.f0)
                .map(f -> f.f1);

        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_keyword_page_view_window", KeywordBean.class));
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}